package march.firstmarch21.march01.plan1;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import march.firstmarch21.march01.CallsBean;

import java.io.*;
import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.HashMap;

public class CallsMapper extends Mapper<LongWritable, Text, CallsBean, NullWritable> {

    String sender;    // 拨号人
    String receiver;  // 接收人
    String p;         // 拨号人居住地
    String c;         // 接受者居住地
    String start;     // 拨号时间
    String end;       // 挂机时间
    String interval;  // 间隔时间
    HashMap<String, String> people = new HashMap();  // 存储人员数据缓存的内容
    HashMap<String, String> provience = new HashMap();  // 存储省份数据缓存的内容
    CallsBean k = new CallsBean();

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        //分别获取缓存文件中的数据
        //  1.获取所有缓存文件集
        URI[] cacheFiles = context.getCacheFiles();
        String line;
        //  2.获取人员文件中的数据
        String path1 = cacheFiles[0].getPath().toString();
        BufferedReader person = new BufferedReader(new InputStreamReader(new FileInputStream(path1),"UTF-8"));
        while (StringUtils.isNotEmpty(line = person.readLine())){
            // 7,18000696806,赵贺彪
            String[] fields = line.split(",");
            people.put(fields[1],fields[2]); // 存储18000696806-->赵贺彪
        }
        //  3.获取省份文件中的数据
        String path2 = cacheFiles[1].getPath().toString();
        BufferedReader city = new BufferedReader(new InputStreamReader(new FileInputStream(path2),"UTF-8"));
        while (StringUtils.isNotEmpty(line = city.readLine())){
            // 1,110000,北京市
            String[] fields = line.split(",");
            provience.put(fields[1],fields[2]); // 存储110000-->北京市
        }
        //  4.关闭资源
        IOUtils.closeStream(person);
        IOUtils.closeStream(city);

    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        // 18620192711,15733218050,1506628174,1506628265,650000,810000 获取一行数据
        String line = value.toString();
        // 拆分
        String[] fields = line.split(",");
        // 替换拨号及接收人名信息\以及他们的所在省份信息
        sender = people.get(fields[0]);
        receiver = people.get(fields[1]);
        p = provience.get(fields[4]);
        c = provience.get(fields[5]);
        // 转换时间日期格式
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy年MM月dd日 HH:mm:ss");
        start = sdf.format(Long.parseLong(fields[2]));
        end = sdf.format(Long.parseLong(fields[3]));
        interval = Long.parseLong(fields[3]) - Long.parseLong(fields[2])+"秒";
        k = new CallsBean(sender,receiver,start,end,interval,p,c);
        System.out.println(k);
        // 写出
        context.write(k,NullWritable.get());
    }
}
